package org.artifact.core.plugin.zookeeper;

import cn.hutool.core.map.MapUtil;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.artifact.core.lang.AbstractPlugin;
import org.artifact.core.server.ServerConfig;

import java.util.Map;
import java.util.Vector;

public class ZooKeeperPlugin extends AbstractPlugin {

	public static final String ADDRESS = "address";
	public static final String SESSION_TIMEOUT = "sessionTimeout";
	public static final String CONNECTION_TIMEOUT = "connectionTimeout";
	private Map config;
	private ZkClient zkClient;
//	ZkSerializer serializer = null;
	// 监听器
	private Vector<StateListener> stateListeners = new Vector<StateListener>();

	public ZooKeeperPlugin() {
		this.config = ServerConfig.me().deepGet("plugins.zookeeper");
	}

	public void addStateListener(StateListener stateListener){
		stateListeners.add(stateListener);
	}

	public void notifyListener(int stat){
		for (StateListener s:stateListeners){
			s.stateChanged(stat);
		}
	}

	@Override
	public boolean start() {
		String address = MapUtil.getStr(config, ADDRESS);
		int session_timeout = MapUtil.getInt(config, SESSION_TIMEOUT);
		int connection_timeout = MapUtil.getInt(config, CONNECTION_TIMEOUT);
		zkClient = new ZkClient(address, session_timeout, connection_timeout);
//		zkClient.subscribeStateChanges(new IZkStateListener() {
//
//			@Override
//			public void handleStateChanged(KeeperState state) throws Exception {
//				System.out.println("----------handleStateChanged = " + state);
//			}
//
//			@Override
//			public void handleSessionEstablishmentError(Throwable error) throws Exception {
//				System.out.println("----------handleSessionEstablishmentError");
//			}
//
//			@Override
//			public void handleNewSession() throws Exception {
//				System.out.println("----------handleNewSession");
//			}
//		});
//		//注册监听
		zkClient.subscribeStateChanges(new IZkStateListener() {
			@Override
			public void handleStateChanged(KeeperState keeperState) throws Exception {
				if (keeperState == Watcher.Event.KeeperState.Disconnected) {
					notifyListener(StateListener.DISCONNECTED);
				} else if (keeperState == Watcher.Event.KeeperState.SyncConnected) {
					notifyListener(StateListener.CONNECTED);
				}
			}

			@Override
			public void handleNewSession() throws Exception {
				notifyListener(StateListener.RECONNECTED);
			}

			@Override
			public void handleSessionEstablishmentError(Throwable throwable) throws Exception {

			}
		});
		return true;
	}

	@Override
	public boolean stop() {
		zkClient.close();
		return true;
	}

	public ZkClient getClient() {
		return zkClient;
	}
}
